package realtime.util;

import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @Create 2023/9/5 18:16
 */
public class EnvUtil {

    /**
     * 获取执行环境
     * @param parallelism 并行度
     * @return
     */
    public static StreamExecutionEnvironment getSEE(Integer parallelism){

        Configuration conf = new Configuration();
        conf.setString("taskmanager.memory.network.max","4096mb");
        conf.setString("taskmanager.memory.network.min","4096mb");
        conf.setDouble("taskmanager.memory.network.fraction",0.4);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.enableCheckpointing(60000, CheckpointingMode.EXACTLY_ONCE);
         env.setParallelism(parallelism);
//        env.getCheckpointConfig().setExternalizedCheckpointCleanup(
//                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
//        );
//        env.getCheckpointConfig().setCheckpointTimeout(60*1000L);
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000L);
//        env.setRestartStrategy(new RestartStrategies.FailureRateRestartStrategyConfiguration(3, Time.days(15),Time.seconds(3)));
//
//        env.setStateBackend(new HashMapStateBackend());
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/edu/ck");
//
//        System.setProperty("HADOOP_USER_NAME","atguigu");

        return env;
    }

}
